// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#pragma once

#include <melon/cportability.h>
#include <melon/random.h>
#include <melon/range.h>
#include <melon/dynamic.h>

#include <cstdint>
#include <cstring>
#include <ctime>
#include <iomanip>
#include <limits>
#include <memory>
#include <optional>
#include <string>
#include <type_traits>
#include <typeindex>
#include <vector>

#include <pollux/common/base/class_name.h>
#include <pollux/common/base/exceptions.h>
#include <pollux/common/serialization/serializable.h>
#include <pollux/type/huge_int.h>
#include <pollux/common/strings/string_view.h>
#include <pollux/type/timestamp.h>
#include <pollux/type/tree.h>

namespace kumo::pollux {
    using int128_t = __int128_t;

    using column_index_t = uint32_t;

    constexpr column_index_t kConstantChannel =
            std::numeric_limits<column_index_t>::max();

    /// Pollux type system supports a small set of SQL-compatible composeable types:
    /// BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, HUGEINT, REAL, DOUBLE, VARCHAR,
    /// VARBINARY, TIMESTAMP, ARRAY, MAP, ROW
    ///
    /// This file has multiple C++ type definitions for each of these logical types.
    /// These logical definitions each serve slightly different purposes.
    /// These type sets are:
    /// - TypeKind
    /// - Type (RowType, BigIntType, ect.)
    /// - Templated Types (Row<T...>, Map<K, V>, ...)
    ///     C++ templated classes. Never instantiated, used to pass limited type
    ///     information into template parameters.

    /// Simple enum with type category.
    enum class TypeKind : int8_t {
        BOOLEAN = 0,
        TINYINT = 1,
        SMALLINT = 2,
        INTEGER = 3,
        BIGINT = 4,
        REAL = 5,
        DOUBLE = 6,
        VARCHAR = 7,
        VARBINARY = 8,
        TIMESTAMP = 9,
        HUGEINT = 10,
        // Enum values for ComplexTypes start after 30 to leave
        // some values space to accommodate adding new scalar/native
        // types above.
        ARRAY = 30,
        MAP = 31,
        ROW = 32,
        UNKNOWN = 33,
        FUNCTION = 34,
        OPAQUE = 35,
        INVALID = 36
    };

    /// Returns the typekind represented by the `name`. Throws if no match found.
    TypeKind mapNameToTypeKind(const std::string &name);

    /// Returns the typekind represented by the `name` and std::nullopt if no
    /// match found.
    std::optional<TypeKind> tryMapNameToTypeKind(const std::string &name);

    std::string mapTypeKindToName(const TypeKind &typeKind);

    std::ostream &operator<<(std::ostream &os, const TypeKind &kind);

    template<TypeKind KIND>
    class ScalarType;
    class ShortDecimalType;
    class LongDecimalType;
    class ArrayType;
    class MapType;
    class RowType;
    class FunctionType;
    class OpaqueType;
    class UnknownType;

    struct UnknownValue {
        bool operator==(const UnknownValue & /* b */) const {
            return true;
        }

        bool operator!=(const UnknownValue & /* b */) const {
            return false;
        }

        bool operator<(const UnknownValue & /* b */) const {
            return false;
        }

        bool operator<=(const UnknownValue & /* b */) const {
            return true;
        }

        bool operator>(const UnknownValue & /* b */) const {
            return false;
        }

        bool operator>=(const UnknownValue & /* b */) const {
            return true;
        }
    };

    template<typename T>
    void toAppend(
        const ::kumo::pollux::UnknownValue & /* value */,
        T * /* result */) {
        // TODO Implement
    }

    template<TypeKind KIND>
    struct TypeTraits {
    };

    template<>
    struct TypeTraits<TypeKind::BOOLEAN> {
        using ImplType = ScalarType<TypeKind::BOOLEAN>;
        using NativeType = bool;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::BOOLEAN;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "BOOLEAN";
    };

    template<>
    struct TypeTraits<TypeKind::TINYINT> {
        using ImplType = ScalarType<TypeKind::TINYINT>;
        using NativeType = int8_t;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::TINYINT;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "TINYINT";
    };

    template<>
    struct TypeTraits<TypeKind::SMALLINT> {
        using ImplType = ScalarType<TypeKind::SMALLINT>;
        using NativeType = int16_t;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::SMALLINT;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "SMALLINT";
    };

    template<>
    struct TypeTraits<TypeKind::INTEGER> {
        using ImplType = ScalarType<TypeKind::INTEGER>;
        using NativeType = int32_t;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::INTEGER;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "INTEGER";
    };

    template<>
    struct TypeTraits<TypeKind::BIGINT> {
        using ImplType = ScalarType<TypeKind::BIGINT>;
        using NativeType = int64_t;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::BIGINT;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "BIGINT";
    };

    template<>
    struct TypeTraits<TypeKind::REAL> {
        using ImplType = ScalarType<TypeKind::REAL>;
        using NativeType = float;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::REAL;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "REAL";
    };

    template<>
    struct TypeTraits<TypeKind::DOUBLE> {
        using ImplType = ScalarType<TypeKind::DOUBLE>;
        using NativeType = double;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::DOUBLE;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "DOUBLE";
    };

    template<>
    struct TypeTraits<TypeKind::VARCHAR> {
        using ImplType = ScalarType<TypeKind::VARCHAR>;
        using NativeType = pollux::StringView;
        using DeepCopiedType = std::string;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::VARCHAR;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "VARCHAR";
    };

    template<>
    struct TypeTraits<TypeKind::TIMESTAMP> {
        using ImplType = ScalarType<TypeKind::TIMESTAMP>;
        using NativeType = Timestamp;
        using DeepCopiedType = Timestamp;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::TIMESTAMP;
        // isPrimitiveType in the type traits indicate whether it is a leaf type.
        // So only types which have other sub types, should be set to false.
        // Timestamp does not contain other types, so it is set to true.
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "TIMESTAMP";
    };

    template<>
    struct TypeTraits<TypeKind::HUGEINT> {
        using ImplType = ScalarType<TypeKind::HUGEINT>;
        using NativeType = int128_t;
        using DeepCopiedType = NativeType;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::HUGEINT;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "HUGEINT";
    };

    template<>
    struct TypeTraits<TypeKind::VARBINARY> {
        using ImplType = ScalarType<TypeKind::VARBINARY>;
        using NativeType = pollux::StringView;
        using DeepCopiedType = std::string;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::VARBINARY;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "VARBINARY";
    };

    template<>
    struct TypeTraits<TypeKind::ARRAY> {
        using ImplType = ArrayType;
        using NativeType = void;
        using DeepCopiedType = void;
        static constexpr uint32_t minSubTypes = 1;
        static constexpr uint32_t maxSubTypes = 1;
        static constexpr TypeKind typeKind = TypeKind::ARRAY;
        static constexpr bool isPrimitiveType = false;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "ARRAY";
    };

    template<>
    struct TypeTraits<TypeKind::MAP> {
        using ImplType = MapType;
        using NativeType = void;
        using DeepCopiedType = void;
        static constexpr uint32_t minSubTypes = 2;
        static constexpr uint32_t maxSubTypes = 2;
        static constexpr TypeKind typeKind = TypeKind::MAP;
        static constexpr bool isPrimitiveType = false;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "MAP";
    };

    template<>
    struct TypeTraits<TypeKind::ROW> {
        using ImplType = RowType;
        using NativeType = void;
        using DeepCopiedType = void;
        static constexpr uint32_t minSubTypes = 1;
        static constexpr uint32_t maxSubTypes = std::numeric_limits<char16_t>::max();
        static constexpr TypeKind typeKind = TypeKind::ROW;
        static constexpr bool isPrimitiveType = false;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "ROW";
    };

    template<>
    struct TypeTraits<TypeKind::UNKNOWN> {
        using ImplType = UnknownType;
        using NativeType = UnknownValue;
        using DeepCopiedType = UnknownValue;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::UNKNOWN;
        static constexpr bool isPrimitiveType = true;
        static constexpr bool isFixedWidth = true;
        static constexpr const char *name = "UNKNOWN";
    };

    template<>
    struct TypeTraits<TypeKind::INVALID> {
        using ImplType = void;
        using NativeType = void;
        using DeepCopiedType = void;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::INVALID;
        static constexpr bool isPrimitiveType = false;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "INVALID";
    };

    template<>
    struct TypeTraits<TypeKind::FUNCTION> {
        using ImplType = FunctionType;
        using NativeType = void;
        using DeepCopiedType = void;
        static constexpr uint32_t minSubTypes = 1;
        static constexpr uint32_t maxSubTypes = std::numeric_limits<char16_t>::max();
        static constexpr TypeKind typeKind = TypeKind::FUNCTION;
        static constexpr bool isPrimitiveType = false;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "FUNCTION";
    };

    template<>
    struct TypeTraits<TypeKind::OPAQUE> {
        using ImplType = OpaqueType;
        using NativeType = std::shared_ptr<void>;
        using DeepCopiedType = std::shared_ptr<void>;
        static constexpr uint32_t minSubTypes = 0;
        static constexpr uint32_t maxSubTypes = 0;
        static constexpr TypeKind typeKind = TypeKind::OPAQUE;
        static constexpr bool isPrimitiveType = false;
        static constexpr bool isFixedWidth = false;
        static constexpr const char *name = "OPAQUE";
    };

    template<TypeKind KIND>
    struct TypeFactory;

#define POLLUX_FLUENT_CAST(NAME, KIND)                                     \
  const typename TypeTraits<TypeKind::KIND>::ImplType& as_##NAME() const { \
    return this->as<TypeKind::KIND>();                                    \
  }                                                                       \
  bool is_##NAME() const {                                                 \
    return this->kind() == TypeKind::KIND;                                \
  }

    class Type;
    using TypePtr = std::shared_ptr<const Type>;

    enum class TypeParameterKind {
        /// Type. For example, element type in the array type.
        kType,
        /// Integer. For example, precision in a decimal type.
        kLongLiteral,
    };

    struct TypeParameter {
        const TypeParameterKind kind;

        /// Must be not not null when kind is kType. All other properties should be
        /// null or unset.
        const TypePtr type;

        /// Must be set when kind is kLongLiteral. All other properties should be null
        /// or unset.
        const std::optional<int64_t> longLiteral;

        /// Creates kType parameter.
        explicit TypeParameter(TypePtr _type)
            : kind{TypeParameterKind::kType},
              type{std::move(_type)},
              longLiteral{std::nullopt} {
        }

        /// Creates kLongLiteral parameter.
        explicit TypeParameter(int64_t _longLiteral)
            : kind{TypeParameterKind::kLongLiteral},
              type{nullptr},
              longLiteral{_longLiteral} {
        }
    };

    /// Abstract class hierarchy. Instances of these classes carry full
    /// information about types, including for example field names.
    /// Can be instantiated by factory methods, like INTEGER()
    /// or MAP(INTEGER(), BIGINT()).
    /// Instances of these classes form a tree, and are immutable.
    /// For example, MAP<INTEGER, ARRAY<BIGINT>> will form a tree like:
    ///
    ///             MapType
    ///           /         \
    ///   IntegerType    ArrayType
    ///                     |
    ///                   BigintType
    class Type : public Tree<const TypePtr>, public pollux::ISerializable {
    public:
        explicit Type(TypeKind kind, bool providesCustomComparison = false)
            : kind_{kind}, providesCustomComparison_(providesCustomComparison) {
        }

        TypeKind kind() const {
            return kind_;
        }

        virtual ~Type() = default;

        /// This convenience method makes pattern matching easier. Rather than having
        /// to know the implementation type up front, just use as<TypeKind::MAP> (for
        /// example) to dynamically cast to the appropriate type.
        template<TypeKind KIND>
        const typename TypeTraits<KIND>::ImplType &as() const {
            return dynamic_cast<const typename TypeTraits<KIND>::ImplType &>(*this);
        }

        virtual bool isPrimitiveType() const = 0;

        /// Returns true if equality relationship is defined for the values of this
        /// type, i.e. a == b is defined and returns true, false or null. For example,
        /// scalar types are usually comparable and complex types are comparable if
        /// their nested types are.
        virtual bool isComparable() const = 0;

        /// Returns true if less than relationship is defined for the values of this
        /// type, i.e. a <= b returns true or false. For example, scalar types are
        /// usually orderable, arrays and structs are orderable if their nested types
        /// are, while map types are not orderable.
        virtual bool isOrderable() const = 0;

        /// Returns true if values of this type implements custom comparison and hash
        /// functions. If this returns true the compare and hash functions in TypeBase
        /// should be used instead of native implementations, e.g. ==, <, >, etc.
        bool providesCustomComparison() const {
            return providesCustomComparison_;
        }

        /// Returns unique logical type name. It can be
        /// different from the physical type name returned by 'kindName()'.
        virtual const char *name() const = 0;

        /// Returns a possibly empty list of type parameters.
        virtual const std::vector<TypeParameter> &parameters() const = 0;

        /// Returns physical type name. Multiple logical types may share the same
        /// physical type backing and therefore return the same physical type name.
        /// The logical type name returned by 'name()' must be unique though.
        virtual const char *kindName() const = 0;

        virtual std::string toString() const = 0;

        /// Options to control the output of to_summary_string().
        struct TypeSummaryOptions {
            /// Maximum number of child types to include in the summary.
            size_type maxChildren{0};
        };

        /// Returns human-readable summary of the type. Useful when full output of
        /// toString() is too large.
        std::string to_summary_string(
            TypeSummaryOptions options = {.maxChildren = 0}) const;

        /// Types are weakly matched.
        /// Examples: Two RowTypes are equivalent if the children types are
        /// equivalent, but the children names could be different. Two OpaqueTypes are
        /// equivalent if the typeKind matches, but the typeIndex could be different.
        virtual bool equivalent(const Type &other) const = 0;

        /// For Complex types (Row, Array, Map, Opaque): types are strongly matched.
        /// For primitive types: same as equivalent.
        virtual bool operator==(const Type &other) const {
            return this->equals(other);
        }

        inline bool operator!=(const Type &other) const {
            return !(*this == other);
        }

        // todo(youknowjack): avoid expensive virtual function calls for these
        // simple functions
        virtual size_t cppSizeInBytes() const {
            // Must be a std::invalid_argument instead of PolluxException in order to
            // generate python ValueError in python bindings.
            throw std::invalid_argument{"Not a fixed width type: " + toString()};
        }

        virtual bool isFixedWidth() const = 0;

        static std::shared_ptr<const Type> create(const melon::Dynamic &obj);

        static void registerSerDe();

        /// Recursive kind hashing (uses only TypeKind).
        size_t hashKind() const;

        /// Recursive kind match (uses only TypeKind).
        bool kindEquals(const std::shared_ptr<const Type> &other) const;

        template<TypeKind KIND, typename... CHILDREN>
        static std::shared_ptr<const typename TypeTraits<KIND>::ImplType> create(
            CHILDREN... children) {
            return TypeFactory<KIND>::create(std::forward<CHILDREN...>(children)...);
        }

        POLLUX_FLUENT_CAST(boolean, BOOLEAN)
        POLLUX_FLUENT_CAST(tinyint, TINYINT)
        POLLUX_FLUENT_CAST(smallint, SMALLINT)
        POLLUX_FLUENT_CAST(integer, INTEGER)
        POLLUX_FLUENT_CAST(bigint, BIGINT)
        POLLUX_FLUENT_CAST(hugeint, HUGEINT)
        POLLUX_FLUENT_CAST(real, REAL)
        POLLUX_FLUENT_CAST(double, DOUBLE)
        POLLUX_FLUENT_CAST(varchar, VARCHAR)
        POLLUX_FLUENT_CAST(varbinary, VARBINARY)
        POLLUX_FLUENT_CAST(timestamp, TIMESTAMP)
        POLLUX_FLUENT_CAST(array, ARRAY)
        POLLUX_FLUENT_CAST(map, MAP)
        POLLUX_FLUENT_CAST(row, ROW)
        POLLUX_FLUENT_CAST(opaque, OPAQUE)
        POLLUX_FLUENT_CAST(unKnown, UNKNOWN)
        POLLUX_FLUENT_CAST(function, FUNCTION)

        const ShortDecimalType &asShortDecimal() const;

        const LongDecimalType &asLongDecimal() const;

        bool isShortDecimal() const;

        bool isLongDecimal() const;

        bool isDecimal() const;

        bool isIntervalYearMonth() const;

        bool isIntervalDayTime() const;

        bool isDate() const;

        bool containsUnknown() const;

    protected:
        MELON_ALWAYS_INLINE bool hasSameTypeId(const Type &other) const {
            return typeid(*this) == typeid(other);
        }

        /// For Complex types (Row, Array, Map, Opaque): types are strongly matched.
        /// Examples: Two RowTypes are == if the children types and the children names
        /// are same. Two OpaqueTypes are == if the typeKind and the typeIndex are
        /// same.
        /// For primitive types: same as equivalent.
        virtual bool equals(const Type &other) const {
            POLLUX_CHECK(this->isPrimitiveType());
            return this->equivalent(other);
        }

    private:
        const TypeKind kind_;
        const bool providesCustomComparison_;

        POLLUX_DEFINE_CLASS_NAME(Type)
    };

#undef POLLUX_FLUENT_CAST

    template<TypeKind KIND, typename = void>
    struct kindCanProvideCustomComparison : std::false_type {
    };

    template<TypeKind KIND>
    struct kindCanProvideCustomComparison<
                KIND,
                std::enable_if_t<
                    TypeTraits<KIND>::isPrimitiveType && TypeTraits<KIND>::isFixedWidth> > {
        static constexpr bool value = true;
    };

    template<TypeKind KIND>
    class TypeBase : public Type {
    public:
        using NativeType = typename TypeTraits<KIND>::NativeType;

        explicit TypeBase(bool providesCustomComparison = false)
            : Type{KIND, providesCustomComparison} {
            if (providesCustomComparison) {
                POLLUX_CHECK(
                    kindCanProvideCustomComparison<KIND>::value,
                    "Custom comparisons are only supported for primitive types that are fixed width.");
            }
        }

        bool isPrimitiveType() const override {
            return TypeTraits<KIND>::isPrimitiveType;
        }

        bool isFixedWidth() const override {
            return TypeTraits<KIND>::isFixedWidth;
        }

        bool isOrderable() const override {
            return false;
        }

        bool isComparable() const override {
            return false;
        }

        const char *kindName() const override {
            return TypeTraits<KIND>::name;
        }

        const char *name() const override {
            return TypeTraits<KIND>::name;
        }

        const std::vector<TypeParameter> &parameters() const override {
            static const std::vector<TypeParameter> kEmpty = {};
            return kEmpty;
        }
    };

    template<TypeKind KIND>
    class CanProvideCustomComparisonType : public TypeBase<KIND> {
    public:
        explicit CanProvideCustomComparisonType(bool providesCustomComparison = false)
            : TypeBase<KIND>{providesCustomComparison} {
        }

        virtual int32_t compare(
            const typename TypeBase<KIND>::NativeType & /*left*/,
            const typename TypeBase<KIND>::NativeType & /*right*/) const {
            POLLUX_CHECK(
                !this->providesCustomComparison(),
                "Type {} is marked as providesCustomComparison but did not implement compare.");
            POLLUX_FAIL("Type {} does not provide custom comparison", this->name());
        }

        virtual uint64_t hash(
            const typename TypeBase<KIND>::NativeType & /*value*/) const {
            POLLUX_CHECK(
                !this->providesCustomComparison(),
                "Type {} is marked as providesCustomComparison but did not implement hash.");
            POLLUX_FAIL("Type {} does not provide custom hash", this->name());
        }
    };

    template<TypeKind KIND>
    class ScalarType : public CanProvideCustomComparisonType<KIND> {
    public:
        explicit ScalarType(bool providesCustomComparison = false)
            : CanProvideCustomComparisonType<KIND>{providesCustomComparison} {
        }

        uint32_t size() const override {
            return 0;
        }

        const std::shared_ptr<const Type> &childAt(uint32_t) const override {
            POLLUX_FAIL("scalar type has no children");
        }

        std::string toString() const override {
            return TypeTraits<KIND>::name;
        }

        bool isOrderable() const override {
            return true;
        }

        bool isComparable() const override {
            return true;
        }

        size_t cppSizeInBytes() const override {
            if (TypeTraits<KIND>::isFixedWidth) {
                return sizeof(typename TypeTraits<KIND>::NativeType);
            }
            // TODO: pollux throws here for non fixed width types.
            return Type::cppSizeInBytes();
        }

        MELON_NOINLINE static const std::shared_ptr<const ScalarType<KIND>> create();

        bool equivalent(const Type &other) const override {
            return Type::hasSameTypeId(other);
        }

        // TODO: pollux implementation is in cpp
        melon::Dynamic serialize() const override {
            melon::Dynamic obj = melon::Dynamic::object;
            obj["name"] = "Type";
            obj["type"] = TypeTraits<KIND>::name;
            return obj;
        }
    };

    template<TypeKind KIND>
    const std::shared_ptr<const ScalarType<KIND>> ScalarType<KIND>::create() {
        static const auto instance = std::make_shared<const ScalarType<KIND>>();
        return instance;
    }

    /// This class represents the fixed-point numbers.
    /// The parameter "precision" represents the number of digits the
    /// Decimal Type can support and "scale" represents the number of digits to
    /// the right of the decimal point.
    template<TypeKind KIND>
    class DecimalType : public ScalarType<KIND> {
    public:
        static_assert(KIND == TypeKind::BIGINT || KIND == TypeKind::HUGEINT);
        static constexpr uint8_t kMaxPrecision = KIND == TypeKind::BIGINT ? 18 : 38;
        static constexpr uint8_t kMinPrecision = KIND == TypeKind::BIGINT ? 1 : 19;

        inline bool equivalent(const Type &other) const override {
            if (!Type::hasSameTypeId(other)) {
                return false;
            }
            const auto &otherDecimal = static_cast<const DecimalType<KIND> &>(other);
            return (
                otherDecimal.precision() == precision() &&
                otherDecimal.scale() == scale());
        }

        inline uint8_t precision() const {
            return parameters_[0].longLiteral.value();
        }

        inline uint8_t scale() const {
            return parameters_[1].longLiteral.value();
        }

        const char *name() const override {
            return "DECIMAL";
        }

        std::string toString() const override {
            return fmt::format("DECIMAL({}, {})", precision(), scale());
        }

        melon::Dynamic serialize() const override {
            auto obj = ScalarType<KIND>::serialize();
            obj["type"] = name();
            obj["precision"] = precision();
            obj["scale"] = scale();
            return obj;
        }

        const std::vector<TypeParameter> &parameters() const override {
            return parameters_;
        }

    protected:
        DecimalType(const uint8_t precision, const uint8_t scale)
            : parameters_{TypeParameter(precision), TypeParameter(scale)} {
            POLLUX_CHECK_LE(
                scale,
                precision,
                "Scale of decimal type must not exceed its precision");
            POLLUX_CHECK_LE(
                precision,
                kMaxPrecision,
                "Precision of decimal type must not exceed {}",
                kMaxPrecision);
            POLLUX_CHECK_GE(
                precision,
                kMinPrecision,
                "Precision of decimal type must be at least {}",
                kMinPrecision);
        }

    private:
        const std::vector<TypeParameter> parameters_;
    };

    class ShortDecimalType : public DecimalType<TypeKind::BIGINT> {
    public:
        ShortDecimalType(int precision, int scale)
            : DecimalType<TypeKind::BIGINT>(precision, scale) {
        }
    };

    class LongDecimalType : public DecimalType<TypeKind::HUGEINT> {
    public:
        LongDecimalType(int precision, int scale)
            : DecimalType<TypeKind::HUGEINT>(precision, scale) {
        }
    };

    TypePtr DECIMAL(uint8_t precision, uint8_t scale);

    MELON_ALWAYS_INLINE const ShortDecimalType &Type::asShortDecimal() const {
        return dynamic_cast<const ShortDecimalType &>(*this);
    }

    MELON_ALWAYS_INLINE const LongDecimalType &Type::asLongDecimal() const {
        return dynamic_cast<const LongDecimalType &>(*this);
    }

    MELON_ALWAYS_INLINE bool Type::isShortDecimal() const {
        return dynamic_cast<const ShortDecimalType *>(this) != nullptr;
    }

    MELON_ALWAYS_INLINE bool Type::isLongDecimal() const {
        return dynamic_cast<const LongDecimalType *>(this) != nullptr;
    }

    MELON_ALWAYS_INLINE bool Type::isDecimal() const {
        return isShortDecimal() || isLongDecimal();
    }

    MELON_ALWAYS_INLINE bool isDecimalName(const std::string &name) {
        return (name == "DECIMAL");
    }

    std::pair<uint8_t, uint8_t> getDecimalPrecisionScale(const Type &type);

    class UnknownType : public CanProvideCustomComparisonType<TypeKind::UNKNOWN> {
    public:
        explicit UnknownType(bool proivdesCustomComparison = false)
            : CanProvideCustomComparisonType<TypeKind::UNKNOWN>(
                proivdesCustomComparison) {
        }

        uint32_t size() const override {
            return 0;
        }

        const std::shared_ptr<const Type> &childAt(uint32_t) const override {
            throw std::invalid_argument{"UnknownType type has no children"};
        }

        std::string toString() const override {
            return TypeTraits<TypeKind::UNKNOWN>::name;
        }

        size_t cppSizeInBytes() const override {
            return 0;
        }

        bool isOrderable() const override {
            return true;
        }

        bool isComparable() const override {
            return true;
        }

        bool equivalent(const Type &other) const override {
            return Type::hasSameTypeId(other);
        }

        melon::Dynamic serialize() const override {
            melon::Dynamic obj = melon::Dynamic::object;
            obj["name"] = "Type";
            obj["type"] = TypeTraits<TypeKind::UNKNOWN>::name;
            return obj;
        }
    };

    class ArrayType : public TypeBase<TypeKind::ARRAY> {
    public:
        explicit ArrayType(TypePtr child);

        explicit ArrayType(
            std::vector<std::string> && /*names*/,
            std::vector<TypePtr> &&types)
            : ArrayType(std::move(types[0])) {
        }

        const TypePtr &elementType() const {
            return child_;
        }

        uint32_t size() const override {
            return 1;
        }

        std::vector<TypePtr> children() const {
            return {child_};
        }

        std::vector<std::string> names() const {
            return {"element"};
        }

        bool isOrderable() const override {
            return child_->isOrderable();
        }

        bool isComparable() const override {
            return child_->isComparable();
        }

        const std::shared_ptr<const Type> &childAt(uint32_t idx) const override;

        const char *nameOf(uint32_t idx) const {
            POLLUX_USER_CHECK_EQ(idx, 0, "Array type should have only one child");
            return "element";
        }

        std::string toString() const override;

        bool equivalent(const Type &other) const override;

        melon::Dynamic serialize() const override;

        const std::vector<TypeParameter> &parameters() const override {
            return parameters_;
        }

    protected:
        bool equals(const Type &other) const override;

        TypePtr child_;
        const std::vector<TypeParameter> parameters_;
    };

    class MapType : public TypeBase<TypeKind::MAP> {
    public:
        MapType(TypePtr keyType, TypePtr valueType);

        explicit MapType(
            std::vector<std::string> && /*names*/,
            std::vector<TypePtr> &&types)
            : MapType(std::move(types[0]), std::move(types[1])) {
        }

        const TypePtr &keyType() const {
            return keyType_;
        }

        const TypePtr &valueType() const {
            return valueType_;
        }

        uint32_t size() const override {
            return 2;
        }

        std::vector<TypePtr> children() const {
            return {keyType_, valueType_};
        }

        std::vector<std::string> names() const {
            return {"key", "value"};
        }

        bool isComparable() const override {
            return keyType_->isComparable() && valueType_->isComparable();
        }

        std::string toString() const override;

        const TypePtr &childAt(uint32_t idx) const override;

        const char *nameOf(uint32_t idx) const;

        bool equivalent(const Type &other) const override;

        melon::Dynamic serialize() const override;

        const std::vector<TypeParameter> &parameters() const override {
            return parameters_;
        }

    protected:
        bool equals(const Type &other) const override;

    private:
        TypePtr keyType_;
        TypePtr valueType_;
        const std::vector<TypeParameter> parameters_;
    };

    class RowType : public TypeBase<TypeKind::ROW> {
    public:
        RowType(
            std::vector<std::string> &&names,
            std::vector<std::shared_ptr<const Type> > &&types);

        ~RowType() override;

        uint32_t size() const final {
            return children_.size();
        }

        const TypePtr &childAt(uint32_t idx) const final {
            POLLUX_CHECK_LT(idx, children_.size());
            return children_[idx];
        }

        const std::vector<std::shared_ptr<const Type> > &children() const {
            return children_;
        }

        bool isOrderable() const override;

        bool isComparable() const override;

        const std::shared_ptr<const Type> &findChild(melon::StringPiece name) const;

        bool containsChild(std::string_view name) const;

        uint32_t getChildIdx(std::string_view name) const;

        std::optional<uint32_t> getChildIdxIfExists(std::string_view name) const;

        const std::string &nameOf(uint32_t idx) const {
            POLLUX_CHECK_LT(idx, names_.size());
            return names_[idx];
        }

        bool equivalent(const Type &other) const override;

        std::string toString() const override;

        /// Print child names and types separated by 'delimiter'.
        void printChildren(std::stringstream &ss, std::string_view delimiter = ",")
        const;

        std::shared_ptr<RowType> unionWith(
            std::shared_ptr<const RowType> rowType) const;

        melon::Dynamic serialize() const override;

        const std::vector<std::string> &names() const {
            return names_;
        }

        const std::vector<TypeParameter> &parameters() const override {
            auto *parameters = parameters_.load();
            if (MELON_UNLIKELY(!parameters)) {
                parameters = makeParameters().release();
                std::vector<TypeParameter> *oldParameters = nullptr;
                if (!parameters_.compare_exchange_strong(oldParameters, parameters)) {
                    delete parameters;
                    parameters = oldParameters;
                }
            }
            return *parameters;
        }

    protected:
        bool equals(const Type &other) const override;

    private:
        std::unique_ptr<std::vector<TypeParameter> > makeParameters() const;

        const std::vector<std::string> names_;
        const std::vector<std::shared_ptr<const Type> > children_;
        mutable std::atomic<std::vector<TypeParameter> *> parameters_{nullptr};
    };

    using RowTypePtr = std::shared_ptr<const RowType>;

    inline RowTypePtr as_row_type(const TypePtr &type) {
        return std::dynamic_pointer_cast<const RowType>(type);
    }

    /// Represents a lambda function. The children are the argument types
    /// followed by the return value type.
    class FunctionType : public TypeBase<TypeKind::FUNCTION> {
    public:
        FunctionType(
            std::vector<std::shared_ptr<const Type> > &&argumentTypes,
            std::shared_ptr<const Type> returnType);

        uint32_t size() const override {
            return children_.size();
        }

        const std::shared_ptr<const Type> &childAt(uint32_t idx) const override {
            POLLUX_CHECK_LT(idx, children_.size());
            return children_[idx];
        }

        const std::vector<std::shared_ptr<const Type> > &children() const {
            return children_;
        }

        bool isOrderable() const override {
            return false;
        }

        bool isComparable() const override {
            return false;
        }

        bool equivalent(const Type &other) const override;

        std::string toString() const override;

        melon::Dynamic serialize() const override;

        const std::vector<TypeParameter> &parameters() const override {
            return parameters_;
        }

    protected:
        bool equals(const Type &other) const override;

    private:
        static std::vector<std::shared_ptr<const Type> > allChildren(
            std::vector<std::shared_ptr<const Type> > &&argumentTypes,
            std::shared_ptr<const Type> returnType) {
            auto children = std::move(argumentTypes);
            children.push_back(returnType);
            return children;
        }

        // Argument types from left to right followed by return value type.
        const std::vector<std::shared_ptr<const Type> > children_;
        const std::vector<TypeParameter> parameters_;
    };

    class OpaqueType : public TypeBase<TypeKind::OPAQUE> {
    public:
        template<typename T>
        using SerializeFunc = std::function<std::string(const std::shared_ptr<T> &)>;
        template<typename T>
        using DeserializeFunc = std::function<std::shared_ptr<T>(const std::string &)>;

        explicit OpaqueType(const std::type_index &typeIndex);

        uint32_t size() const override {
            return 0;
        }

        const std::shared_ptr<const Type> &childAt(uint32_t) const override {
            POLLUX_FAIL("OpaqueType type has no children");
        }

        std::string toString() const override;

        bool equivalent(const Type &other) const override;

        const std::type_index &typeIndex() const {
            return typeIndex_;
        }

        melon::Dynamic serialize() const override;

        /// In special cases specific OpaqueTypes might want to serialize additional
        /// metadata. In those cases we need to deserialize it back. Since
        /// OpaqueType::create<T>() returns canonical type for T without metadata,
        /// we allow to create new instance here or return nullptr if the same one
        /// can be used. Note that it's about deserialization of type itself,
        /// DeserializeFunc above is about deserializing instances of the type. It's
        /// implemented as a virtual member instead of a standalone registry just
        /// for convenience.
        virtual std::shared_ptr<const OpaqueType> deserializeExtra(
            const melon::Dynamic &json) const;

        /// Function for converting std::shared_ptr<T> into a string. Always returns
        /// non-nullptr function or throws if not function has been registered.
        SerializeFunc<void> getSerializeFunc() const;

        DeserializeFunc<void> getDeserializeFunc() const;

        template<typename Class>
        MELON_NOINLINE static std::shared_ptr<const OpaqueType> create() {
            /// static vars in templates are dangerous across DSOs, but it's just a
            /// performance optimization. Comparison looks at type_index anyway.
            static const auto instance =
                    std::make_shared<const OpaqueType>(std::type_index(typeid(Class)));
            return instance;
        }

        /// This function currently doesn't do synchronization neither with reads
        /// or writes, so it's caller's responsibility to not invoke it concurrently
        /// with other Pollux code. Usually it'd be invoked at static initialization
        /// time. It can be changed in the future if it becomes a problem.
        template<typename T>
        MELON_NOINLINE static void registerSerialization(
            const std::string &persistentName,
            SerializeFunc<T> serialize = nullptr,
            DeserializeFunc<T> deserialize = nullptr) {
            SerializeFunc<void> serializeTypeErased;
            if (serialize) {
                serializeTypeErased =
                        [serialize](const std::shared_ptr<void> &x) -> std::string {
                            return serialize(std::static_pointer_cast<T>(x));
                        };
            }
            DeserializeFunc<void> deserializeTypeErased;
            if (deserialize) {
                deserializeTypeErased =
                        [deserialize](const std::string &s) -> std::shared_ptr<void> {
                            return std::static_pointer_cast<void>(deserialize(s));
                        };
            }
            registerSerializationTypeErased(
                OpaqueType::create<T>(),
                persistentName,
                serializeTypeErased,
                deserializeTypeErased);
        }

        static void clearSerializationRegistry();

    protected:
        bool equals(const Type &other) const override;

    private:
        const std::type_index typeIndex_;

        static void registerSerializationTypeErased(
            const std::shared_ptr<const OpaqueType> &type,
            const std::string &persistentName,
            SerializeFunc<void> serialize = nullptr,
            DeserializeFunc<void> deserialize = nullptr);
    };

    using IntegerType = ScalarType<TypeKind::INTEGER>;
    using BooleanType = ScalarType<TypeKind::BOOLEAN>;
    using TinyintType = ScalarType<TypeKind::TINYINT>;
    using SmallintType = ScalarType<TypeKind::SMALLINT>;
    using BigintType = ScalarType<TypeKind::BIGINT>;
    using HugeintType = ScalarType<TypeKind::HUGEINT>;
    using RealType = ScalarType<TypeKind::REAL>;
    using DoubleType = ScalarType<TypeKind::DOUBLE>;
    using TimestampType = ScalarType<TypeKind::TIMESTAMP>;
    using VarcharType = ScalarType<TypeKind::VARCHAR>;
    using VarbinaryType = ScalarType<TypeKind::VARBINARY>;

    constexpr long kMillisInSecond = 1000;
    constexpr long kMillisInMinute = 60 * kMillisInSecond;
    constexpr long kMillisInHour = 60 * kMillisInMinute;
    constexpr long kMillisInDay = 24 * kMillisInHour;

    /// Time interval in milliseconds.
    class IntervalDayTimeType : public BigintType {
    private:
        IntervalDayTimeType() = default;

    public:
        static const std::shared_ptr<const IntervalDayTimeType> &get() {
            static const std::shared_ptr<const IntervalDayTimeType> kType{
                new IntervalDayTimeType()
            };
            return kType;
        }

        const char *name() const override {
            return "INTERVAL DAY TO SECOND";
        }

        bool equivalent(const Type &other) const override {
            // Pointer comparison works since this type is a singleton.
            return this == &other;
        }

        std::string toString() const override {
            return name();
        }

        /// Returns the interval 'value' (milliseconds) formatted as DAYS
        /// HOURS:MINUTES:SECONDS.MILLIS. For example, 1 03:48:20.100.
        /// TODO Figure out how to make this API generic, i.e. available via Type.
        /// Perhaps, Type::valueToString(variant)?
        std::string valueToString(int64_t value) const;

        melon::Dynamic serialize() const override {
            melon::Dynamic obj = melon::Dynamic::object;
            obj["name"] = "IntervalDayTimeType";
            obj["type"] = name();
            return obj;
        }

        static TypePtr deserialize(const melon::Dynamic & /*obj*/) {
            return IntervalDayTimeType::get();
        }
    };

    MELON_ALWAYS_INLINE const std::shared_ptr<const IntervalDayTimeType> &
    INTERVAL_DAY_TIME() {
        return IntervalDayTimeType::get();
    }

    MELON_ALWAYS_INLINE bool Type::isIntervalDayTime() const {
        // Pointer comparison works since this type is a singleton.
        return (this == INTERVAL_DAY_TIME().get());
    }

    constexpr long kMonthInYear = 12;

    /// Time interval in months.
    class IntervalYearMonthType : public IntegerType {
    private:
        IntervalYearMonthType() = default;

    public:
        static const std::shared_ptr<const IntervalYearMonthType> &get() {
            static const std::shared_ptr<const IntervalYearMonthType> kType{
                new IntervalYearMonthType()
            };
            return kType;
        }

        const char *name() const override {
            return "INTERVAL YEAR TO MONTH";
        }

        bool equivalent(const Type &other) const override {
            // Pointer comparison works since this type is a singleton.
            return this == &other;
        }

        std::string toString() const override {
            return name();
        }

        /// Returns the interval 'value' (months) formatted as YEARS MONTHS.
        /// For example, 14 months (INTERVAL '1-2' YEAR TO MONTH) would be
        /// represented as 1-2; -14 months would be represents as -1-2.
        std::string valueToString(int32_t value) const;

        melon::Dynamic serialize() const override {
            melon::Dynamic obj = melon::Dynamic::object;
            obj["name"] = "IntervalYearMonthType";
            obj["type"] = name();
            return obj;
        }

        static TypePtr deserialize(const melon::Dynamic & /*obj*/) {
            return IntervalYearMonthType::get();
        }
    };

    MELON_ALWAYS_INLINE const std::shared_ptr<const IntervalYearMonthType> &
    INTERVAL_YEAR_MONTH() {
        return IntervalYearMonthType::get();
    }

    MELON_ALWAYS_INLINE bool Type::isIntervalYearMonth() const {
        // Pointer comparison works since this type is a singleton.
        return (this == INTERVAL_YEAR_MONTH().get());
    }

    /// Date is represented as the number of days since epoch start using int32_t.
    class DateType : public IntegerType {
    private:
        DateType() = default;

    public:
        static const std::shared_ptr<const DateType> &get() {
            static const std::shared_ptr<const DateType> kType{new DateType()};
            return kType;
        }

        const char *name() const override {
            return "DATE";
        }

        bool equivalent(const Type &other) const override {
            return this == &other;
        }

        std::string toString() const override {
            return name();
        }

        std::string toString(int32_t days) const;

        /// Returns a date, represented as days since epoch,
        /// as an ISO 8601-formatted string.
        static std::string toIso8601(int32_t days);

        int32_t toDays(melon::StringPiece in) const;

        int32_t toDays(const char *in, size_t len) const;

        melon::Dynamic serialize() const override {
            melon::Dynamic obj = melon::Dynamic::object;
            obj["name"] = "DateType";
            obj["type"] = name();
            return obj;
        }

        static TypePtr deserialize(const melon::Dynamic & /*obj*/) {
            return DateType::get();
        }
    };

    MELON_ALWAYS_INLINE const std::shared_ptr<const DateType> &DATE() {
        return DateType::get();
    }

    MELON_ALWAYS_INLINE bool isDateName(const std::string &name) {
        return (name == DateType::get()->name());
    }

    MELON_ALWAYS_INLINE bool Type::isDate() const {
        // The pointers can be compared since DATE is a singleton.
        return this == DATE().get();
    }

    /// Used as T for SimpleVector subclasses that wrap another vector when
    /// the wrapped vector is of a complex type. Applies to
    /// DictionaryVector, SequenceVector and ConstantVector. This must have
    /// a size different from any of the scalar data type sizes to enable
    /// run time checking with 'elementSize_'.
    struct ComplexType {
        TypePtr create() {
            POLLUX_NYI();
        }

        operator melon::Dynamic() const {
            return melon::Dynamic("ComplexType");
        }
    };

    template<TypeKind KIND>
    struct TypeFactory {
        static std::shared_ptr<const typename TypeTraits<KIND>::ImplType> create() {
            return TypeTraits<KIND>::ImplType::create();
        }
    };

    template<>
    struct TypeFactory<TypeKind::UNKNOWN> {
        static std::shared_ptr<const UnknownType> create() {
            return std::make_shared<UnknownType>();
        }
    };

    template<>
    struct TypeFactory<TypeKind::ARRAY> {
        static std::shared_ptr<const ArrayType> create(TypePtr elementType) {
            return std::make_shared<ArrayType>(std::move(elementType));
        }
    };

    template<>
    struct TypeFactory<TypeKind::MAP> {
        static std::shared_ptr<const MapType> create(
            TypePtr keyType,
            TypePtr valType) {
            return std::make_shared<MapType>(std::move(keyType), std::move(valType));
        }
    };

    template<>
    struct TypeFactory<TypeKind::ROW> {
        static std::shared_ptr<const RowType> create(
            std::vector<std::string> &&names,
            std::vector<TypePtr> &&types) {
            return std::make_shared<const RowType>(std::move(names), std::move(types));
        }
    };

    std::shared_ptr<const ArrayType> ARRAY(TypePtr elementType);

    std::shared_ptr<const RowType> ROW(
        std::vector<std::string> &&names,
        std::vector<TypePtr> &&types);

    std::shared_ptr<const RowType> ROW(
        std::initializer_list<std::pair<const std::string, TypePtr> > &&pairs);

    std::shared_ptr<const RowType> ROW(std::vector<TypePtr> &&pairs);

    std::shared_ptr<const MapType> MAP(TypePtr keyType, TypePtr valType);

    std::shared_ptr<const FunctionType> FUNCTION(
        std::vector<TypePtr> &&argumentTypes,
        TypePtr returnType);

    template<typename Class>
    std::shared_ptr<const OpaqueType> OPAQUE() {
        return OpaqueType::create<Class>();
    }

#define POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(TEMPLATE_FUNC, typeKind, ...)      \
  [&]() {                                                                     \
    switch (typeKind) {                                                       \
      case ::kumo::pollux::TypeKind::BOOLEAN: {                            \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::BOOLEAN>(           \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::INTEGER: {                            \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::INTEGER>(           \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::TINYINT: {                            \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::TINYINT>(           \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::SMALLINT: {                           \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::SMALLINT>(          \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::BIGINT: {                             \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::BIGINT>(            \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::HUGEINT: {                            \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::HUGEINT>(           \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::REAL: {                               \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::REAL>(__VA_ARGS__); \
      }                                                                       \
      case ::kumo::pollux::TypeKind::DOUBLE: {                             \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::DOUBLE>(            \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::VARCHAR: {                            \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::VARCHAR>(           \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::VARBINARY: {                          \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::VARBINARY>(         \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::TIMESTAMP: {                          \
        return TEMPLATE_FUNC<::kumo::pollux::TypeKind::TIMESTAMP>(         \
            __VA_ARGS__);                                                     \
      }                                                                       \
      default:                                                                \
        POLLUX_FAIL(                                                           \
            "not a scalar type! kind: {}", mapTypeKindToName(typeKind));      \
    }                                                                         \
  }()

#define POLLUX_DYNAMIC_SCALAR_TEMPLATE_TYPE_DISPATCH(                     \
    TEMPLATE_FUNC, T, typeKind, ...)                                     \
  [&]() {                                                                \
    switch (typeKind) {                                                  \
      case ::kumo::pollux::TypeKind::BOOLEAN: {                       \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::BOOLEAN>(   \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::INTEGER: {                       \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::INTEGER>(   \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::TINYINT: {                       \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::TINYINT>(   \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::SMALLINT: {                      \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::SMALLINT>(  \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::BIGINT: {                        \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::BIGINT>(    \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::HUGEINT: {                       \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::HUGEINT>(   \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::REAL: {                          \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::REAL>(      \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::DOUBLE: {                        \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::DOUBLE>(    \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::VARCHAR: {                       \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::VARCHAR>(   \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::VARBINARY: {                     \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::VARBINARY>( \
            __VA_ARGS__);                                                \
      }                                                                  \
      case ::kumo::pollux::TypeKind::TIMESTAMP: {                     \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::TIMESTAMP>( \
            __VA_ARGS__);                                                \
      }                                                                  \
      default:                                                           \
        POLLUX_FAIL(                                                      \
            "not a scalar type! kind: {}", mapTypeKindToName(typeKind)); \
    }                                                                    \
  }()

#define POLLUX_DYNAMIC_TEMPLATE_TYPE_DISPATCH(TEMPLATE_FUNC, T, typeKind, ...) \
  [&]() {                                                                     \
    switch (typeKind) {                                                       \
      case ::kumo::pollux::TypeKind::BOOLEAN: {                            \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::BOOLEAN>(        \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::INTEGER: {                            \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::INTEGER>(        \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::TINYINT: {                            \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::TINYINT>(        \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::SMALLINT: {                           \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::SMALLINT>(       \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::BIGINT: {                             \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::BIGINT>(         \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::HUGEINT: {                            \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::HUGEINT>(        \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::REAL: {                               \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::REAL>(           \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::DOUBLE: {                             \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::DOUBLE>(         \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::VARCHAR: {                            \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::VARCHAR>(        \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::VARBINARY: {                          \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::VARBINARY>(      \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::TIMESTAMP: {                          \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::TIMESTAMP>(      \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::MAP: {                                \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::MAP>(            \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::ARRAY: {                              \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::ARRAY>(          \
            __VA_ARGS__);                                                     \
      }                                                                       \
      case ::kumo::pollux::TypeKind::ROW: {                                \
        return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::ROW>(            \
            __VA_ARGS__);                                                     \
      }                                                                       \
      default:                                                                \
        POLLUX_FAIL("not a known type kind: {}", mapTypeKindToName(typeKind)); \
    }                                                                         \
  }()

#define POLLUX_DYNAMIC_TEMPLATE_TYPE_DISPATCH_ALL(                    \
    TEMPLATE_FUNC, T, typeKind, ...)                                 \
  [&]() {                                                            \
    if ((typeKind) == ::kumo::pollux::TypeKind::UNKNOWN) {        \
      return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::UNKNOWN>( \
          __VA_ARGS__);                                              \
    } else if ((typeKind) == ::kumo::pollux::TypeKind::OPAQUE) {  \
      return TEMPLATE_FUNC<T, ::kumo::pollux::TypeKind::OPAQUE>(  \
          __VA_ARGS__);                                              \
    } else {                                                         \
      return POLLUX_DYNAMIC_TEMPLATE_TYPE_DISPATCH(                   \
          TEMPLATE_FUNC, T, typeKind, __VA_ARGS__);                  \
    }                                                                \
  }()

#define POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH_ALL(TEMPLATE_FUNC, typeKind, ...)   \
  [&]() {                                                                      \
    if ((typeKind) == ::kumo::pollux::TypeKind::UNKNOWN) {                  \
      return TEMPLATE_FUNC<::kumo::pollux::TypeKind::UNKNOWN>(__VA_ARGS__); \
    } else if ((typeKind) == ::kumo::pollux::TypeKind::OPAQUE) {            \
      return TEMPLATE_FUNC<::kumo::pollux::TypeKind::OPAQUE>(__VA_ARGS__);  \
    } else {                                                                   \
      return POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(                               \
          TEMPLATE_FUNC, typeKind, __VA_ARGS__);                               \
    }                                                                          \
  }()

#define POLLUX_DYNAMIC_TYPE_DISPATCH_IMPL(PREFIX, SUFFIX, typeKind, ...)        \
  [&]() {                                                                      \
    switch (typeKind) {                                                        \
      case ::kumo::pollux::TypeKind::BOOLEAN: {                             \
        return PREFIX<::kumo::pollux::TypeKind::BOOLEAN> SUFFIX(            \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::INTEGER: {                             \
        return PREFIX<::kumo::pollux::TypeKind::INTEGER> SUFFIX(            \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::TINYINT: {                             \
        return PREFIX<::kumo::pollux::TypeKind::TINYINT> SUFFIX(            \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::SMALLINT: {                            \
        return PREFIX<::kumo::pollux::TypeKind::SMALLINT> SUFFIX(           \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::BIGINT: {                              \
        return PREFIX<::kumo::pollux::TypeKind::BIGINT> SUFFIX(             \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::HUGEINT: {                             \
        return PREFIX<::kumo::pollux::TypeKind::HUGEINT> SUFFIX(            \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::REAL: {                                \
        return PREFIX<::kumo::pollux::TypeKind::REAL> SUFFIX(__VA_ARGS__);  \
      }                                                                        \
      case ::kumo::pollux::TypeKind::DOUBLE: {                              \
        return PREFIX<::kumo::pollux::TypeKind::DOUBLE> SUFFIX(             \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::VARCHAR: {                             \
        return PREFIX<::kumo::pollux::TypeKind::VARCHAR> SUFFIX(            \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::VARBINARY: {                           \
        return PREFIX<::kumo::pollux::TypeKind::VARBINARY> SUFFIX(          \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::TIMESTAMP: {                           \
        return PREFIX<::kumo::pollux::TypeKind::TIMESTAMP> SUFFIX(          \
            __VA_ARGS__);                                                      \
      }                                                                        \
      case ::kumo::pollux::TypeKind::ARRAY: {                               \
        return PREFIX<::kumo::pollux::TypeKind::ARRAY> SUFFIX(__VA_ARGS__); \
      }                                                                        \
      case ::kumo::pollux::TypeKind::MAP: {                                 \
        return PREFIX<::kumo::pollux::TypeKind::MAP> SUFFIX(__VA_ARGS__);   \
      }                                                                        \
      case ::kumo::pollux::TypeKind::ROW: {                                 \
        return PREFIX<::kumo::pollux::TypeKind::ROW> SUFFIX(__VA_ARGS__);   \
      }                                                                        \
      default:                                                                 \
        POLLUX_FAIL("not a known type kind: {}", mapTypeKindToName(typeKind));  \
    }                                                                          \
  }()

#define POLLUX_DYNAMIC_TYPE_DISPATCH(TEMPLATE_FUNC, typeKind, ...) \
  POLLUX_DYNAMIC_TYPE_DISPATCH_IMPL(TEMPLATE_FUNC, , typeKind, __VA_ARGS__)

#define POLLUX_DYNAMIC_TYPE_DISPATCH_ALL(TEMPLATE_FUNC, typeKind, ...)          \
  [&]() {                                                                      \
    if ((typeKind) == ::kumo::pollux::TypeKind::UNKNOWN) {                  \
      return TEMPLATE_FUNC<::kumo::pollux::TypeKind::UNKNOWN>(__VA_ARGS__); \
    } else if ((typeKind) == ::kumo::pollux::TypeKind::OPAQUE) {            \
      return TEMPLATE_FUNC<::kumo::pollux::TypeKind::OPAQUE>(__VA_ARGS__);  \
    } else {                                                                   \
      return POLLUX_DYNAMIC_TYPE_DISPATCH_IMPL(                                 \
          TEMPLATE_FUNC, , typeKind, __VA_ARGS__);                             \
    }                                                                          \
  }()

#define POLLUX_DYNAMIC_TYPE_DISPATCH_METHOD( \
    CLASS_NAME, METHOD_NAME, typeKind, ...) \
  POLLUX_DYNAMIC_TYPE_DISPATCH_IMPL(         \
      CLASS_NAME, ::METHOD_NAME, typeKind, __VA_ARGS__)

#define POLLUX_DYNAMIC_TYPE_DISPATCH_METHOD_ALL(                             \
    CLASS_NAME, METHOD_NAME, typeKind, ...)                                 \
  [&]() {                                                                   \
    if ((typeKind) == ::kumo::pollux::TypeKind::UNKNOWN) {               \
      return CLASS_NAME<::kumo::pollux::TypeKind::UNKNOWN>::METHOD_NAME( \
          __VA_ARGS__);                                                     \
    } else if ((typeKind) == ::kumo::pollux::TypeKind::OPAQUE) {         \
      return CLASS_NAME<::kumo::pollux::TypeKind::OPAQUE>::METHOD_NAME(  \
          __VA_ARGS__);                                                     \
    } else {                                                                \
      return POLLUX_DYNAMIC_TYPE_DISPATCH_IMPL(                              \
          CLASS_NAME, ::METHOD_NAME, typeKind, __VA_ARGS__);                \
    }                                                                       \
  }()

#define POLLUX_SCALAR_ACCESSOR(KIND) \
  std::shared_ptr<const ScalarType<TypeKind::KIND>> KIND()

#define POLLUX_STATIC_FIELD_DYNAMIC_DISPATCH(CLASS, FIELD, typeKind)           \
  [&]() {                                                                     \
    switch (typeKind) {                                                       \
      case ::kumo::pollux::TypeKind::BOOLEAN: {                            \
        return CLASS<::kumo::pollux::TypeKind::BOOLEAN>::FIELD;            \
      }                                                                       \
      case ::kumo::pollux::TypeKind::INTEGER: {                            \
        return CLASS<::kumo::pollux::TypeKind::INTEGER>::FIELD;            \
      }                                                                       \
      case ::kumo::pollux::TypeKind::TINYINT: {                            \
        return CLASS<::kumo::pollux::TypeKind::TINYINT>::FIELD;            \
      }                                                                       \
      case ::kumo::pollux::TypeKind::SMALLINT: {                           \
        return CLASS<::kumo::pollux::TypeKind::SMALLINT>::FIELD;           \
      }                                                                       \
      case ::kumo::pollux::TypeKind::BIGINT: {                             \
        return CLASS<::kumo::pollux::TypeKind::BIGINT>::FIELD;             \
      }                                                                       \
      case ::kumo::pollux::TypeKind::REAL: {                               \
        return CLASS<::kumo::pollux::TypeKind::REAL>::FIELD;               \
      }                                                                       \
      case ::kumo::pollux::TypeKind::DOUBLE: {                             \
        return CLASS<::kumo::pollux::TypeKind::DOUBLE>::FIELD;             \
      }                                                                       \
      case ::kumo::pollux::TypeKind::VARCHAR: {                            \
        return CLASS<::kumo::pollux::TypeKind::VARCHAR>::FIELD;            \
      }                                                                       \
      case ::kumo::pollux::TypeKind::VARBINARY: {                          \
        return CLASS<::kumo::pollux::TypeKind::VARBINARY>::FIELD;          \
      }                                                                       \
      case ::kumo::pollux::TypeKind::TIMESTAMP: {                          \
        return CLASS<::kumo::pollux::TypeKind::TIMESTAMP>::FIELD;          \
      }                                                                       \
      case ::kumo::pollux::TypeKind::ARRAY: {                              \
        return CLASS<::kumo::pollux::TypeKind::ARRAY>::FIELD;              \
      }                                                                       \
      case ::kumo::pollux::TypeKind::MAP: {                                \
        return CLASS<::kumo::pollux::TypeKind::MAP>::FIELD;                \
      }                                                                       \
      case ::kumo::pollux::TypeKind::ROW: {                                \
        return CLASS<::kumo::pollux::TypeKind::ROW>::FIELD;                \
      }                                                                       \
      default:                                                                \
        POLLUX_FAIL("not a known type kind: {}", mapTypeKindToName(typeKind)); \
    }                                                                         \
  }()

#define POLLUX_STATIC_FIELD_DYNAMIC_DISPATCH_ALL(CLASS, FIELD, typeKind)   \
  [&]() {                                                                 \
    if ((typeKind) == ::kumo::pollux::TypeKind::UNKNOWN) {             \
      return CLASS<::kumo::pollux::TypeKind::UNKNOWN>::FIELD;          \
    } else if ((typeKind) == ::kumo::pollux::TypeKind::OPAQUE) {       \
      return CLASS<::kumo::pollux::TypeKind::OPAQUE>::FIELD;           \
    } else if ((typeKind) == ::kumo::pollux::TypeKind::HUGEINT) {      \
      return CLASS<::kumo::pollux::TypeKind::HUGEINT>::FIELD;          \
    } else {                                                              \
      return POLLUX_STATIC_FIELD_DYNAMIC_DISPATCH(CLASS, FIELD, typeKind); \
    }                                                                     \
  }()

    // todo: union convenience creators

    POLLUX_SCALAR_ACCESSOR(INTEGER);

    POLLUX_SCALAR_ACCESSOR(BOOLEAN);

    POLLUX_SCALAR_ACCESSOR(TINYINT);

    POLLUX_SCALAR_ACCESSOR(SMALLINT);

    POLLUX_SCALAR_ACCESSOR(BIGINT);

    POLLUX_SCALAR_ACCESSOR(HUGEINT);

    POLLUX_SCALAR_ACCESSOR(REAL);

    POLLUX_SCALAR_ACCESSOR(DOUBLE);

    POLLUX_SCALAR_ACCESSOR(TIMESTAMP);

    POLLUX_SCALAR_ACCESSOR(VARCHAR);

    POLLUX_SCALAR_ACCESSOR(VARBINARY);

    TypePtr UNKNOWN();

    template<TypeKind KIND>
    std::shared_ptr<const Type> createScalarType() {
        return ScalarType<KIND>::create();
    }

    std::shared_ptr<const Type> createScalarType(TypeKind kind);

    std::shared_ptr<const Type> createType(
        TypeKind kind,
        std::vector<std::shared_ptr<const Type> > &&children);

    /// Returns true built-in or custom type with specified name exists.
    bool hasType(const std::string &name);

    /// Returns built-in or custom type with specified name and child types.
    /// Returns nullptr if type with specified name doesn't exist.
    TypePtr getType(
        const std::string &name,
        const std::vector<TypeParameter> &parameters);

    template<TypeKind KIND>
    std::shared_ptr<const Type> createType(
        std::vector<std::shared_ptr<const Type> > &&children) {
        if (children.size() != 0) {
            throw std::invalid_argument{
                std::string(TypeTraits<KIND>::name) +
                " primitive type takes no children"
            };
        }
        static_assert(TypeTraits<KIND>::isPrimitiveType);
        return ScalarType<KIND>::create();
    }

    template<>
    std::shared_ptr<const Type> createType<TypeKind::ROW>(
        std::vector<std::shared_ptr<const Type> > &&children);

    template<>
    std::shared_ptr<const Type> createType<TypeKind::ARRAY>(
        std::vector<std::shared_ptr<const Type> > &&children);

    template<>
    std::shared_ptr<const Type> createType<TypeKind::MAP>(
        std::vector<std::shared_ptr<const Type> > &&children);

    template<>
    std::shared_ptr<const Type> createType<TypeKind::OPAQUE>(
        std::vector<std::shared_ptr<const Type> > &&children);

#undef POLLUX_SCALAR_ACCESSOR

    template<typename T>
    struct SimpleTypeTrait {
    };

    template<>
    struct SimpleTypeTrait<int128_t> : public TypeTraits<TypeKind::HUGEINT> {
    };

    template<>
    struct SimpleTypeTrait<int64_t> : public TypeTraits<TypeKind::BIGINT> {
    };

    template<>
    struct SimpleTypeTrait<int32_t> : public TypeTraits<TypeKind::INTEGER> {
    };

    template<>
    struct SimpleTypeTrait<int16_t> : public TypeTraits<TypeKind::SMALLINT> {
    };

    template<>
    struct SimpleTypeTrait<int8_t> : public TypeTraits<TypeKind::TINYINT> {
    };

    template<>
    struct SimpleTypeTrait<float> : public TypeTraits<TypeKind::REAL> {
    };

    template<>
    struct SimpleTypeTrait<double> : public TypeTraits<TypeKind::DOUBLE> {
    };

    template<>
    struct SimpleTypeTrait<bool> : public TypeTraits<TypeKind::BOOLEAN> {
    };

    template<>
    struct SimpleTypeTrait<Timestamp> : public TypeTraits<TypeKind::TIMESTAMP> {
    };

    template<>
    struct SimpleTypeTrait<UnknownValue> : public TypeTraits<TypeKind::UNKNOWN> {
    };

    template<TypeKind KIND>
    static inline int32_t sizeOfTypeKindHelper() {
        return sizeof(typename TypeTraits<KIND>::NativeType);
    }

    static inline int32_t sizeOfTypeKind(TypeKind kind) {
        if (kind == TypeKind::BOOLEAN) {
            throw std::invalid_argument("sizeOfTypeKind dos not apply to boolean");
        }
        return POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(sizeOfTypeKindHelper, kind);
    }

    template<typename T, typename U>
    static inline T to(const U &value) {
        return melon::to<T>(value);
    }

    template<>
    inline Timestamp to(const std::string &value) {
        return Timestamp(0, 0);
    }

    template<>
    inline UnknownValue to(const std::string & /* value */) {
        return UnknownValue();
    }

    template<>
    inline std::string to(const Timestamp &value) {
        return value.toString();
    }

    template<>
    inline std::string to(const int128_t &value) {
        return std::to_string(value);
    }

    template<>
    inline std::string to(const pollux::StringView &value) {
        return std::string(value.data(), value.size());
    }

    template<>
    inline std::string to(const ComplexType &value) {
        return std::string("ComplexType");
    }

    template<>
    inline pollux::StringView to(const std::string &value) {
        return pollux::StringView(value);
    }

    namespace exec {
        /// Forward declaration.
        class CastOperator;

        using CastOperatorPtr = std::shared_ptr<const CastOperator>;
    } // namespace exec

    /// Forward declaration.
    class variant;
    class AbstractInputGenerator;

    using AbstractInputGeneratorPtr = std::shared_ptr<AbstractInputGenerator>;
    using FuzzerGenerator = melon::detail::DefaultGenerator;

    struct InputGeneratorConfig {
        // TODO: hook up the rest options in VectorFuzzer::Options.
        size_t seed_;
        double nullRatio_;
    };

    /// Associates custom types with their custom operators to be the payload in
    /// the custom type registry.
    class CustomTypeFactories {
    public:
        virtual ~CustomTypeFactories();

        /// Returns a shared pointer to the custom type.
        virtual TypePtr getType(
            const std::vector<TypeParameter> &parameters) const = 0;

        /// Returns a shared pointer to the custom cast operator. If a custom type
        /// should be treated as its underlying native type during type castings,
        /// return a nullptr. If a custom type does not support castings, throw an
        /// exception.
        virtual exec::CastOperatorPtr getCastOperator() const = 0;

        virtual AbstractInputGeneratorPtr getInputGenerator(
            const InputGeneratorConfig &config) const = 0;
    };

    class AbstractInputGenerator {
    public:
        AbstractInputGenerator(
            size_t seed,
            const TypePtr &type,
            std::unique_ptr<AbstractInputGenerator> &&next,
            double nullRatio)
            : type_{type}, next_{std::move(next)}, nullRatio_{nullRatio} {
            rng_.seed(seed);
        }

        virtual ~AbstractInputGenerator();

        virtual variant generate() = 0;

        TypePtr type() const {
            return type_;
        }

    protected:
        FuzzerGenerator rng_;

        TypePtr type_;

        std::unique_ptr<AbstractInputGenerator> next_;

        double nullRatio_;
    };

    /// Adds custom type to the registry if it doesn't exist already. No-op if
    /// type with specified name already exists. Returns true if type was added,
    /// false if type with the specified name already exists.
    bool registerCustomType(
        const std::string &name,
        std::unique_ptr<const CustomTypeFactories> factories);

    // See registerOpaqueType() for documentation on type index and opaque type
    // alias.
    melon::F14FastMap<std::string, std::type_index> &getTypeIndexByOpaqueAlias();

    // Reverse of getTypeIndexByOpaqueAlias() when we need to look up the opaque
    // alias by its type index.
    melon::F14FastMap<std::type_index, std::string> &getOpaqueAliasByTypeIndex();

    std::type_index getTypeIdForOpaqueTypeAlias(const std::string &name);

    std::string getOpaqueAliasForTypeId(std::type_index typeIndex);

    /// OpaqueType represents a type that is not part of the Pollux type system.
    /// To identify the underlying type we use std::type_index which is stable
    /// within the same process. However, it is not necessarily stable across
    /// processes.
    ///
    /// So if we were to serialize an opaque type using its std::type_index, we
    /// might not be able to deserialize it in another process. To solve this
    /// problem, we require that both the serializing and deserializing processes
    /// register the opaque type using registerOpaqueType() with the same alias.
    template<typename Class>
    bool registerOpaqueType(const std::string &alias) {
        auto typeIndex = std::type_index(typeid(Class));
        return getTypeIndexByOpaqueAlias().emplace(alias, typeIndex).second &&
               getOpaqueAliasByTypeIndex().emplace(typeIndex, alias).second;
    }

    /// Unregisters an opaque type. Returns true if the type was unregistered.
    /// Currently, it is only used for testing to provide isolation between tests
    /// when using the same alias.
    template<typename Class>
    bool unregisterOpaqueType(const std::string &alias) {
        auto typeIndex = std::type_index(typeid(Class));
        return getTypeIndexByOpaqueAlias().erase(alias) == 1 &&
               getOpaqueAliasByTypeIndex().erase(typeIndex) == 1;
    }

    /// Return true if a custom type with the specified name exists.
    bool customTypeExists(const std::string &name);

    /// Returns a set of all registered custom type names.
    std::unordered_set<std::string> getCustomTypeNames();

    /// Returns an instance of a custom type with the specified name and specified
    /// child types.
    TypePtr getCustomType(
        const std::string &name,
        const std::vector<TypeParameter> &parameters);

    /// Removes custom type from the registry if exists. Returns true if type was
    /// removed, false if type didn't exist.
    bool unregisterCustomType(const std::string &name);

    /// Returns the custom cast operator for the custom type with the specified
    /// name. Returns nullptr if a type with the specified name does not exist or
    /// does not have a dedicated custom cast operator.
    exec::CastOperatorPtr getCustomTypeCastOperator(const std::string &name);

    /// Returns the input generator for the custom type with the specified name.
    AbstractInputGeneratorPtr getCustomTypeInputGenerator(
        const std::string &name,
        const InputGeneratorConfig &config);

    // Allows us to transparently use melon::toAppend(), melon::join(), etc.
    template<class TString>
    void toAppend(
        const std::shared_ptr<const kumo::pollux::Type> &type,
        TString *result) {
        result->append(type->toString());
    }

    /// Appends type's SQL string to 'out'. Uses DuckDB SQL.
    void toTypeSql(const TypePtr &type, std::ostream &out);

    /// Cache of serialized RowType instances. Useful to reduce the size of
    /// serialized expressions and plans. Disabled by default. Not thread safe.
    ///
    /// To enable, call 'serializedTypeCache().enable()'. This enables the cache for
    /// the current thread. To disable, call 'serializedTypeCache().disable()'.
    /// While enables, type serialization will use the cache and serialize the types
    /// using IDs stored in the cache. The caller is responsible for saving
    /// serialized types from the cache and using these to hidrate
    /// 'deserializedTypeCache()' before deserializing the types.
    class SerializedTypeCache {
    public:
        struct Options {
            // Caching applies to RowType's with at least this many fields.
            size_t minRowTypeSize = 10;
        };

        bool isEnabled() const {
            return enabled_;
        }

        const Options &options() const {
            return options_;
        }

        void enable(const Options &options = {.minRowTypeSize = 10}) {
            enabled_ = true;
            options_ = options;
        }

        void disable() {
            enabled_ = false;
        }

        size_t size() const {
            return cache_.size();
        }

        void clear() {
            cache_.clear();
        }

        /// Returns the ID of the type if it is in the cache. Returns std::nullopt if
        /// type is not found in the cache. Cache key is type instance pointer. Hence,
        /// equal but different instances are stored separately.
        std::optional<int32_t> get(const Type &type) const;

        /// Stores the type in the cache. Returns the ID of the type. Reports an error
        /// if type is already present in the cache. IDs are monotonically increasing.
        /// Serialized type may refer to types stored previously in the cache. When
        /// deserializing type cache, make sure to deserialize types in the order of
        /// cache IDs.
        int32_t put(const Type &type, melon::Dynamic serialized);

        /// Serialized the types stored in the cache. Use
        /// DeserializedTypeCache::deserialize to deserialize.
        melon::Dynamic serialize();

    private:
        bool enabled_{false};
        Options options_;
        melon::F14FastMap<const Type *, std::pair<int32_t, melon::Dynamic> > cache_;
    };

    /// Thread local cache of serialized RowType instances. Used by
    /// RowType::serialize.
    SerializedTypeCache &serializedTypeCache();

    /// Thread local cache of deserialized RowType instances. Used when
    /// deserializing Type objects.
    class DeserializedTypeCache {
    public:
        void deserialize(const melon::Dynamic &obj);

        size_t size() const {
            return cache_.size();
        }

        const TypePtr &get(int32_t id) const;

        void clear() {
            cache_.clear();
        }

    private:
        melon::F14FastMap<int32_t, TypePtr> cache_;
    };

    DeserializedTypeCache &deserializedTypeCache();
} // namespace kumo::pollux

namespace melon {
    template<>
    struct hasher<::kumo::pollux::UnknownValue> {
        size_t operator()(const ::kumo::pollux::UnknownValue /* value */) const {
            return 0;
        }
    };

    // Helper functions to allow TypeKind and some common variations to be
    // transparently used by melon::sformat.
    //
    // e.g: melon::sformat("type: {}", typeKind);
    template<>
    class FormatValue<kumo::pollux::TypeKind> {
    public:
        explicit FormatValue(const kumo::pollux::TypeKind &type) : type_(type) {
        }

        template<typename FormatCallback>
        void format(FormatArg &arg, FormatCallback &cb) const {
            return format_value::formatString(
                kumo::pollux::mapTypeKindToName(type_), arg, cb);
        }

    private:
        kumo::pollux::TypeKind type_;
    };

    /// Prints all types derived from `pollux::Type`.
    template<typename T>
    class FormatValue<
                std::shared_ptr<T>,
                typename std::enable_if_t<std::is_base_of_v<kumo::pollux::Type, T> > > {
    public:
        explicit FormatValue(const std::shared_ptr<const kumo::pollux::Type> &type)
            : type_(type) {
        }

        template<typename FormatCallback>
        void format(FormatArg &arg, FormatCallback &cb) const {
            return format_value::formatString(type_->toString(), arg, cb);
        }

    private:
        std::shared_ptr<const kumo::pollux::Type> type_;
    };
} // namespace melon

template<>
struct fmt::formatter<kumo::pollux::TypeKind> : fmt::formatter<string_view> {
    template<typename FormatContext>
    auto format(kumo::pollux::TypeKind k, FormatContext &ctx) const {
        return formatter<string_view>::format(
            kumo::pollux::mapTypeKindToName(k), ctx);
    }
};

template<typename T>
struct fmt::formatter<
            std::shared_ptr<T>,
            typename std::
            enable_if_t<std::is_base_of_v<kumo::pollux::Type, T>, char> >
        : fmt::formatter<string_view> {
    template<typename FormatContext>
    auto format(const std::shared_ptr<T> &k, FormatContext &ctx) const {
        return formatter<string_view>::format(k->toString(), ctx);
    }
};
